package app

import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, run}
import com.cw.realtime.common.bean.TrafficHomeDetailPageViewBean
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.DateFormatUtil._
import com.cw.realtime.common.util.FlinkSinkUtil
import function.KPFunc4HomeDetail
import io.circe._
import io.circe.parser._
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import io.circe.generic.auto._
import io.circe.syntax._

import java.time.Duration

object DwsTrafficHomeDetailPageView {

  def main(args: Array[String]): Unit = {

    implicit val config = RunnerConfig("dws_traffic_home_detail_page_view_window", TOPIC_DWD_TRAFFIC_PAGE, 10023, ckp = true)
    run { (_, ds) =>

      // 清洗数据
      val jsonObjectStream = etl(ds)
      //      jsonObjectStream.map(_.toJson.noSpaces).print()

      // 统计首页、详情页的的独立访客数
      val pageViewStream = uvCount(jsonObjectStream)
      //      pageViewStream.print()

      // 设定水位线
      val pageViewStreamWM = withWatermark(pageViewStream)

      // 开窗聚合
      val aggStream = windowAgg(pageViewStreamWM)

      aggStream
        .map(_.asJson.noSpaces)
        .sinkTo(FlinkSinkUtil.getDorisSink(DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW))


    }

    def etl(ds: DataStream[String]) = {
      ds.flatMap { s =>
        for {
          json <- parse(s).toOption
          jsonObj <- json.asObject
          page <- jsonObj("page")
          common <- jsonObj("common")
          _ <- common.asObject.get("mid")
          pageId <- page.asObject.get("page_id")
          pageId <- pageId.asString if pageId == "home" || pageId == "good_detail"
        } yield jsonObj
      }
    }


  }

  def uvCount(ds: DataStream[JsonObject]) = {
    ds.keyBy(_("common").get.asObject.get("mid").get.asString.get)
      .process(new KPFunc4HomeDetail())
  }

  def withWatermark(ds: DataStream[TrafficHomeDetailPageViewBean]) = {
    ds.assignTimestampsAndWatermarks(
      WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(
        new SerializableTimestampAssigner[TrafficHomeDetailPageViewBean] {
          override def extractTimestamp(element: TrafficHomeDetailPageViewBean, recordTimestamp: Long): Long =
            element.ts
        }))
  }

  def windowAgg(ds: DataStream[TrafficHomeDetailPageViewBean]) = {
    ds.windowAll(TumblingEventTimeWindows.of(time.Time.seconds(10)))
      .reduce(_ + _, (win: TimeWindow, iter: Iterable[TrafficHomeDetailPageViewBean], collector: Collector[TrafficHomeDetailPageViewBean]) => {
        val stt = win.getStart.toDateTime
        val edt = win.getEnd.toDateTime
        val curDt = System.currentTimeMillis.toDate
        iter.foreach(pv => collector.collect(pv.transWindowRange(stt, edt, curDt)))
      })
  }
}
